package com.demo.webflux.jpaadapter;

import org.reactivestreams.Publisher;
import org.springframework.data.repository.CrudRepository;
import org.springframework.data.repository.reactive.ReactiveCrudRepository;

import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Scheduler;

// 实现了ReactiveCrudRepository，并具有与代理存储库相同的泛型类型
public abstract class ReactiveCrudRepositoryAdapter<T, ID, I extends CrudRepository<T, ID>> implements ReactiveCrudRepository<T, ID> {

	protected final I delegate;
	protected final Scheduler scheduler;

	public ReactiveCrudRepositoryAdapter(I delegate, Scheduler scheduler) {
		super();
		this.delegate = delegate;
		this.scheduler = scheduler;
	}

	// 阻塞式save方法的响应式包装方法，阻塞调用被包装到Mono.fromCallable操作符中并转移到专用的scheduler
	@Override
	public <S extends T> Mono<S> save(S entity){
		return Mono.fromCallable(() -> delegate.save(entity))
				.subscribeOn(scheduler);
	}

	// findById的响应式适配器，首先订阅流id，如果值到达，则调用delegate实例。CrudRepository的findById方法返回Optional，因此需要将值映射到Mono实例。一旦收到一个空的Optional
	// 则返回空的Mono，执行将被转移到专用的scheduler
	@Override
	public Mono<T> findById(Publisher<ID> id) {
		return Mono.from(id).flatMap(actualId -> delegate.findById(actualId)
				.map(Mono::just)
				.orElseGet(Mono::empty))
				.subscribeOn(scheduler);
	}

	// deleteAll的响应式适配器，由于deleteAll(Publisher<? extends T> entities)和deleteAll(Iterator<? extends T> entities)方法具有不同的语意，我们无法将一个响应式调用直接映射到
	// 一个阻塞式调用。例如，实体流是无限的，因此不会删除任何项目。因此，deleteAll方法订阅实体并为每个实体发出单独的delegate.delete(entity)请求。由于删除请求可以并行执行，因此每个
	// 请求都有自己的subscribeOn调用，以从scheduler接收工作单元。deleteAll方法返回一个输出流，该输出流在传入流终止且所有删除操作完成时完成。
	@Override
	public Mono<Void> deleteAll(Publisher<? extends T> entities) {
		return Flux.from(entities)
				.flatMap(entity -> Mono.fromRunnable(() -> delegate.delete(entity))
						.subscribeOn(scheduler))
				.then();
	}

	@Override
	public <S extends T> Flux<S> saveAll(Iterable<S> entities) {
		// TODO Auto-generated method stub
		return null;
	}

	@Override
	public <S extends T> Flux<S> saveAll(Publisher<S> entityStream) {
		// TODO Auto-generated method stub
		return null;
	}

	@Override
	public Mono<T> findById(ID id) {
		// TODO Auto-generated method stub
		return null;
	}

	@Override
	public Mono<Boolean> existsById(ID id) {
		// TODO Auto-generated method stub
		return null;
	}

	@Override
	public Mono<Boolean> existsById(Publisher<ID> id) {
		// TODO Auto-generated method stub
		return null;
	}

	@Override
	public Flux<T> findAll() {
		// TODO Auto-generated method stub
		return null;
	}

	@Override
	public Flux<T> findAllById(Iterable<ID> ids) {
		// TODO Auto-generated method stub
		return null;
	}

	@Override
	public Flux<T> findAllById(Publisher<ID> idStream) {
		// TODO Auto-generated method stub
		return null;
	}

	@Override
	public Mono<Long> count() {
		// TODO Auto-generated method stub
		return null;
	}

	@Override
	public Mono<Void> deleteById(ID id) {
		// TODO Auto-generated method stub
		return null;
	}

	@Override
	public Mono<Void> deleteById(Publisher<ID> id) {
		// TODO Auto-generated method stub
		return null;
	}

	@Override
	public Mono<Void> delete(T entity) {
		// TODO Auto-generated method stub
		return null;
	}

	@Override
	public Mono<Void> deleteAll(Iterable<? extends T> entities) {
		// TODO Auto-generated method stub
		return null;
	}

	@Override
	public Mono<Void> deleteAll() {
		// TODO Auto-generated method stub
		return null;
	}

}
